package cn.doitedu.demo5;

import cn.doitedu.demo1.UserEvent_1;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/20
 * @Desc: 学大数据，上多易教育
 * <p>
 * 支持多个规则
 * 规则1： 只要用户发生了X行为，就命中规则
 * 规则2： 只要用户发生了W行为，就命中规则
 **/
public class Demo5 {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 配置一些必选参数
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");


        // 构建一个kafka source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setTopics("user-events-log")
                .setBootstrapServers("doitedu:9092")
                .setGroupId("g001")
                //.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 将source添加到env，得到数据流
        DataStreamSource<String> jsonStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s1");

        // json解析，封装成javaBean
        SingleOutputStreamOperator<UserEvent_1> beanStream = jsonStream.map(value -> JSON.parseObject(value, UserEvent_1.class));

        // 将数据keyBy(用户)
        KeyedStream<UserEvent_1, Long> keyByStream = beanStream.keyBy(UserEvent_1::getUid);

        // 处理逻辑
        keyByStream.process(
                new KeyedProcessFunction<Long, UserEvent_1, String>() {

            JSONObject message = new JSONObject();
            HashMap<String, RuleCalculator> ruleCalculatorPool = new HashMap<>();


            @Override
            public void open(Configuration parameters) throws Exception {

                // 构造规则1的运算机对象
                RuleCalculator rule1Calculator = new Rule1_Calculator();
                // 初始化规则1的运算机对象
                rule1Calculator.init(getRuntimeContext());

                // 构造规则2的运算机对象
                RuleCalculator rule2Calculator = new Rule2_Calculator();
                rule2Calculator.init(getRuntimeContext());


                // 构造规则3的运算机对象
                RuleCalculator rule3Calculator = new Rule3_Calculator();
                rule3Calculator.init(getRuntimeContext());


                // 将运算机对象放入运算机池
                ruleCalculatorPool.put("rule-001", rule1Calculator);
                ruleCalculatorPool.put("rule-002", rule2Calculator);
                ruleCalculatorPool.put("rule-003", rule3Calculator);

            }

            @Override
            public void processElement(UserEvent_1 eventBean, KeyedProcessFunction<Long, UserEvent_1, String>.Context ctx, Collector<String> out) throws Exception {

                // 遍历运算机池中的每一个规则运算机，来处理当前收到的用户行为
                for (Map.Entry<String, RuleCalculator> entry : ruleCalculatorPool.entrySet()) {
                    RuleCalculator ruleCalculator = entry.getValue();
                    // 调用该规则运算机，处理本次收到的行为事件
                    ruleCalculator.calc(eventBean, out);
                }

            }
        });


    }
}
